package com.google.code.coma.rmi.storage;

import com.google.code.coma.rmi.EntrySelector;
import com.google.code.coma.rmi.ObserveCallback;
import com.google.code.coma.rmi.RMICallback;
import com.google.code.coma.rmi.SelectionException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.rmi.RemoteException;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author Bernhard Nickel, e0925384@student.tuwien.ac.at
 * @author Dominik Strasser, dominikstr@gmail.com
 */
public class StorageServiceImpl<T extends Serializable> implements StorageService<T> {
	private static final Log log = LogFactory.getLog(StorageServiceImpl.class);

	private List<T> entries = Collections.synchronizedList(new ArrayList<T>());
	private List<EntrySelector<T>> entrySelectors = new ArrayList<EntrySelector<T>>();
	private Map<EntrySelector<T>, RMICallback<T>> callbacks = new HashMap<EntrySelector<T>, RMICallback<T>>();
	private Map<ObserveCallback.ChangeAction, List<ObserveCallback<T>>> observeCallbacks = new HashMap<ObserveCallback.ChangeAction, List<ObserveCallback<T>>>();

	private ThreadPoolExecutor executor;

	public StorageServiceImpl() {
		executor = new ThreadPoolExecutor(5, 10, 10, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>());
	}

	@Override
	public void add(final T entry) throws RemoteException {
		executor.execute(new Runnable() {
			@Override
			public void run() {
				addEntry(entry);
				executeCallback();
			}
		});
	}

	@Override
	public void update(T entry) throws RemoteException {

		try {
			Field f = entry.getClass().getDeclaredField("id");
			f.setAccessible(true);

			for (int i = 0; i < entries.size(); i++) {
				Serializable knownEntry = entries.get(i);
				if (entry.getClass().isInstance(knownEntry)) {
					if (f.get(entry).equals(f.get(knownEntry))) {
						entries.set(i, entry);
						executeObserveCallback(ObserveCallback.ChangeAction.UPDATE);
						return;
					}
				}
			}
		} catch (Exception e) {
			throw new RemoteException("Error updating entry", e);
		}
	}

	@Override
	public List<T> getSnapshot(EntrySelector<T> selector) throws RemoteException {
		try {
			return selector.selectEntries(new ArrayList<T>(entries));
		} catch (SelectionException e) {
			if (log.isDebugEnabled()) {
				log.debug("Keine ahnung");
			}
		}

		return null;
	}

	@Override
	public List<T> take(EntrySelector<T> selector) throws RemoteException {
		try {
			List<T> selectedEntries = selector.selectEntries(new ArrayList<T>(entries));
			removeEntries(selectedEntries);
			return selectedEntries;
		} catch (SelectionException e) {
			if (log.isDebugEnabled()) {
				log.debug("Skipped selector " + selector.getClass());
			}
		}

		return new ArrayList<T>();
	}

	private void executeCallback() {
		EntrySelector<T> acceptedSelector = null;

		for (int i = 0, entrySelectorsSize = entrySelectors.size(); i < entrySelectorsSize; i++) {
			EntrySelector<T> selector;
			synchronized (this) {
				if (i > entrySelectors.size()) {
					break;
				}
				selector = entrySelectors.get(i);
			}
			try {
				List<T> selectedEntries = selector.selectEntries(new ArrayList<T>(entries));
				acceptedSelector = selector;
				removeEntries(selectedEntries);
				try {
					callbacks.get(acceptedSelector).handle(selectedEntries);
					break;
				} catch (Exception e) {
					synchronized (this) {
						entries.addAll(selectedEntries);
					}
					acceptedSelector = null;
				}
			} catch (SelectionException e) {
				//Nothing to do here
				if (log.isDebugEnabled()) {
					log.debug("Skipped selector " + selector.getClass());
				}
			}
		}


		if (acceptedSelector != null) {
			entrySelectors.remove(acceptedSelector);
			entrySelectors.add(acceptedSelector);
		}
	}

	@SuppressWarnings("unchecked")
	private void addEntry(T entry) {
		try {
			//Convention over configuration for the RMI part (this is different from the space implementation but we wanted
			//to implement both concepts)
			for (Field f : entry.getClass().getDeclaredFields()) {
				f.setAccessible(true);
				if (f.getName().equals("id") && f.get(entry) == null) {
					f.set(entry, UUID.randomUUID().toString());
					break;
				}
			}
		} catch (Exception e) {
			log.error("Unable to create unique id", e);
		}
		synchronized (this) {
			entries.add(entry);
		}

		executeObserveCallback(ObserveCallback.ChangeAction.ADD);
	}

	@SuppressWarnings("unchecked")
	private void removeEntries(List<T> entries) {
		synchronized (this) {
			this.entries.removeAll(entries);
		}
		executeObserveCallback(ObserveCallback.ChangeAction.REMOVE);
	}

	private void executeObserveCallback(ObserveCallback.ChangeAction changeAction) {
		List<ObserveCallback<T>> addObserveCallbacks = observeCallbacks.get(changeAction);

		if (addObserveCallbacks == null) {
			return;
		}

		for (int i = 0; i < addObserveCallbacks.size(); i++) {
			ObserveCallback observeCallback = addObserveCallbacks.get(i);
			try {
				observeCallback.onChange(changeAction, new ArrayList<T>(entries));
			} catch (Exception e) {
				log.warn("Error calling observer", e);
			}
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public synchronized void registerCallback(EntrySelector<T> selector, RMICallback<T> callback) throws RemoteException {

		entrySelectors.add(selector);

		callbacks.put(selector, callback);
		if (log.isInfoEnabled()) {
			log.info("Registered new callback");
		}

		executeCallback();
	}

	@Override
	public synchronized void registerObserveCallback(ObserveCallback<T> observeCallback, ObserveCallback.ChangeAction... changeActions) throws RemoteException {

		for (ObserveCallback.ChangeAction changeAction : changeActions) {
			List<ObserveCallback<T>> actionObserveCallbacks = observeCallbacks.get(changeAction);

			if (actionObserveCallbacks == null) {
				actionObserveCallbacks = new ArrayList<ObserveCallback<T>>();
				observeCallbacks.put(changeAction, actionObserveCallbacks);
			}
			actionObserveCallbacks.add(observeCallback);
			if (log.isInfoEnabled()) {
				log.info("Registered new " + changeAction + " observer");
			}
		}
	}
}
